﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Reactive;
using System.Reactive.Linq;
using Microsoft.WindowsAzure.StorageClient;
using System.Data.Services.Client;
using System.Diagnostics.Contracts;

namespace Microsoft.WindowsAzure.StorageClient
{
    /// <summary>
    /// Extension methods for CloudQueue.
    /// </summary>
    public static class CloudTableExtensions
    {
        /// <summary>
        /// Saves changes asynchronous.
        /// </summary>
        /// <param name="context">Table service context.</param>
        /// <returns>Asynchronous operation.</returns>
        public static IObservable<DataServiceResponse> SaveChangesAsync(this TableServiceContext context)
        {
            Contract.Requires(context != null);

            var saveChanges = Observable.FromAsyncPattern<DataServiceResponse>(context.BeginSaveChanges, context.EndSaveChanges);
            return Observable.Defer(saveChanges);
        }

        /// <summary>
        /// Saves changes asynchronous.
        /// </summary>
        /// <param name="context">Table service context.</param>
        /// <param name="options">The options for how the client can save the pending set of changes.</param>
        /// <returns>Asynchronous operation.</returns>
        public static IObservable<DataServiceResponse> SaveChangesAsync(this TableServiceContext context, SaveChangesOptions options)
        {
            Contract.Requires(context != null);
            var saveChanges = Observable.FromAsyncPattern<SaveChangesOptions, DataServiceResponse>(context.BeginSaveChanges, context.EndSaveChanges);
            return Observable.Defer(() => saveChanges(options));
        }

        /// <summary>
        /// Saves changes with retries asynchronous.
        /// </summary>
        /// <param name="context">Table service context.</param>
        /// <returns>Asynchronous operation.</returns>
        public static IObservable<DataServiceResponse> SaveChangesWithRetriesAsync(this TableServiceContext context)
        {
            Contract.Requires(context != null);
            var saveChanges = Observable.FromAsyncPattern<DataServiceResponse>(context.BeginSaveChangesWithRetries, context.EndSaveChangesWithRetries);
            return Observable.Defer(saveChanges);
        }

        /// <summary>
        /// Saves changes with retries asynchronous.
        /// </summary>
        /// <param name="context">Table service context.</param>
        /// <param name="options">The options for how the client can save the pending set of changes.</param>
        /// <returns>Asynchronous operation.</returns>
        public static IObservable<DataServiceResponse> SaveChangesWithRetriesAsync(this TableServiceContext context, SaveChangesOptions options)
        {
            Contract.Requires(context != null);
            var saveChanges = Observable.FromAsyncPattern<SaveChangesOptions, DataServiceResponse>(context.BeginSaveChangesWithRetries, context.EndSaveChangesWithRetries);
            return Observable.Defer(() => saveChanges(options));
        }

        /// <summary>
        /// Executes query asynchronous.
        /// </summary>
        /// <param name="query">Query.</param>
        /// <returns>Asynchronous operation.</returns>
        public static IObservable<ResultSegment<T>> ExecuteAsync<T>(this CloudTableQuery<T> query)
        {
            Contract.Requires(query != null);

            var execute = Observable.FromAsyncPattern<ResultSegment<T>>(query.BeginExecuteSegmented, query.EndExecuteSegmented);
            return Observable.Defer(execute);
        }

        /// <summary>
        /// Executes query with continuation asynchronous.
        /// </summary>
        /// <param name="query">Query.</param>
        /// <param name="continuationToken">A continuation token returned by a previous listing operation.</param>
        /// <returns>Asynchronous operation.</returns>
        public static IObservable<ResultSegment<T>> ExecuteAsync<T>(this CloudTableQuery<T> query, ResultContinuation continuationToken)
        {
            Contract.Requires(query != null);
            Contract.Requires(continuationToken != null);

            var execute = Observable.FromAsyncPattern<ResultContinuation, ResultSegment<T>>(query.BeginExecuteSegmented, query.EndExecuteSegmented);
            return Observable.Defer(() => execute(continuationToken));
        }

        /// <summary>
        /// Gets the next result segment asynchronous.
        /// </summary>
        /// <typeparam name="T">Result segment.</typeparam>
        /// <param name="segment">Next result segment.</param>
        /// <returns>Asynchronous operation.</returns>
        public static IObservable<ResultSegment<T>> GetNextAsync<T>(this ResultSegment<T> segment)
        {
            Contract.Requires(segment != null);

            var getNext = Observable.FromAsyncPattern<ResultSegment<T>>(segment.BeginGetNext, segment.EndGetNext);
            return Observable.Defer(getNext);
        }

        /// <summary>
        /// Returns all elements for query asynchronous.
        /// </summary>
        /// <param name="query">Query.</param>
        /// <returns>Asynchronous operation.</returns>
        public static IObservable<T> GetAllAsync<T>(this CloudTableQuery<T> query)
        {
            Contract.Requires(query != null);

            return query.ExecuteAsync()
                         .Expand(r => Observable.If(() => r.HasMoreResults,r.GetNextAsync()))
                         .SelectMany(r => r.Results);
        }
    }
}
